06 错误处理与日志

Agent API 上线后，可能会遇到大模型调用超时、数据库连接失败、参数异常等问题。如果后端直接返回 Flask 默认的 HTML 错误页面，前端在按 JSON 解析时就可能出现异常。

对于 API 服务，错误响应也应保持统一的 JSON 格式，而不是返回默认 HTML 错误页面。

同时，服务需要记录日志，包括出错请求、错误原因、用户标识、调用链路等信息。日志是线上问题定位的重要依据。

一、默认错误行为

Flask 内置了一套 HTTP 异常：400（请求错误）、401（未授权）、403（禁止访问）、404（未找到）、405（方法不允许）、500（服务器错误）等。

默认情况下，这些异常会返回 HTML 错误页面。对于 API 服务，需要将其转换为 JSON 格式。

二、注册错误处理器

可以使用@app.errorhandler()装饰器注册自定义错误处理函数：

python

from flask import Flask, jsonify

app = Flask(__name__)


@app.errorhandler(400)
def bad_request(error):
    return jsonify({"error": str(error.description)}), 400


@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "接口不存在"}), 404


@app.errorhandler(405)
def method_not_allowed(error):
    return jsonify({"error": "请求方法不允许"}), 405


@app.errorhandler(500)
def internal_error(error):
    return jsonify({"error": "服务器内部错误"}), 500

注册后，错误响应会变为 JSON：

json

{"error": "接口不存在"}

注意：错误处理函数需要手动返回状态码（第二个返回值），否则 Flask 会默认返回 200。

2.1 按异常类注册

除了使用状态码，也可以按异常类注册：

python

from werkzeug.exceptions import BadRequest, NotFound

@app.errorhandler(BadRequest)
def handle_bad_request(error):
    return jsonify({"error": str(error.description)}), 400

@app.errorhandler(NotFound)
def handle_not_found(error):
    return jsonify({"error": "接口不存在"}), 404

状态码和异常类可以对应使用，例如BadRequest.code == 400。

2.2 捕获所有HTTP异常

如果需要统一处理所有 HTTP 错误，可以注册HTTPException：

python

from flask import json
from werkzeug.exceptions import HTTPException

@app.errorhandler(HTTPException)
def handle_http_exception(error):
    """所有HTTP错误统一返回JSON"""
    response = error.get_response()
    response.data = json.dumps({
        "code": error.code,
        "name": error.name,
        "description": error.description,
    })
    response.content_type = "application/json"
    return response

这种方式可以将所有 HTTP 错误统一转换为 JSON。

2.3 捕获未知异常

除了 HTTP 异常，代码中还可能出现数据库连接失败、第三方 API 超时、空值处理异常等非 HTTP 异常。可以再注册一个Exception处理器：

python

@app.errorhandler(Exception)
def handle_exception(error):
    # HTTP异常走已注册的处理器
    if isinstance(error, HTTPException):
        return error

    # 其他异常：记录日志，返回500
    app.logger.error(f"未处理的异常: {error}", exc_info=True)
    return jsonify({"error": "服务器内部错误"}), 500

关键点是先判断异常是否为HTTPException。如果是，则直接返回，让其继续走 HTTP 错误处理流程；只有非 HTTP 异常才按未知异常处理。

三、自定义异常类

Agent API 中经常需要表达更明确的业务错误，例如参数缺失、会话不存在、模型调用失败等。此时可以定义自定义异常类，携带错误码和额外信息：

python

from flask import jsonify


class AgentError(Exception):
    """Agent API的基础异常"""
    status_code = 400

    def __init__(self, message, status_code=None, payload=None):
        super().__init__()
        self.message = message
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload

    def to_dict(self):
        rv = dict(self.payload or ())
        rv["error"] = self.message
        rv["code"] = self.status_code
        return rv

再为该异常注册处理器：

python

@app.errorhandler(AgentError)
def handle_agent_error(error):
    return jsonify(error.to_dict()), error.status_code

业务代码中可以直接抛出该异常：

python

@app.route("/chat", methods=["POST"])
def chat():
    data = request.get_json()
    if not data:
        raise AgentError("请求体不能为空")

    message = data.get("message")
    if not message:
        raise AgentError("缺少message字段")

    session_id = data.get("session_id")
    if session_id and len(session_id) > 100:
        raise AgentError(
            "session_id过长",
            status_code=400,
            payload={"max_length": 100},
        )

    return {"reply": f"收到: {message}"}

返回的错误格式如下：

json

{
    "error": "session_id过长",
    "code": 400,
    "max_length": 100
}

四、abort函数

如果只需要快速返回一个 HTTP 错误，可以使用abort函数：

python

from flask import abort

@app.route("/chat", methods=["POST"])
def chat():
    data = request.get_json()
    if not data or "message" not in data:
        abort(400, description="缺少message字段")

    return {"reply": "收到"}

abort(400)会立即停止当前函数，抛出一个BadRequest异常，然后交给已注册的错误处理器。

description参数会变成error.description，错误处理器中可以通过str(error.description)获取。

五、Blueprint错误处理

Blueprint 也可以定义自己的错误处理器：

python

chat_bp = Blueprint("chat", __name__)

@chat_bp.errorhandler(429)
def rate_limit_exceeded(error):
    return jsonify({"error": "请求太频繁，请稍后再试"}), 429

Blueprint 的错误处理器只会在该蓝图的视图函数中触发。如果蓝图中没有匹配的处理器，会继续查找应用级别的处理器。

一种常见做法是按路径前缀区分错误格式：

python

@app.errorhandler(404)
@app.errorhandler(405)
def handle_api_error(error):
    if request.path.startswith("/api/"):
        return jsonify({"error": str(error.description)}), error.code
    else:
        return error  # 非API路径返回默认HTML

六、日志基础

Flask 使用 Python 标准库中的logging模块，可以通过app.logger访问：

python

app.logger.debug("调试信息")
app.logger.info("一般信息")
app.logger.warning("警告信息")
app.logger.error("错误信息")
app.logger.critical("严重错误")

这五个级别从低到高为：DEBUG < INFO < WARNING < ERROR < CRITICAL。低于当前配置级别的日志会被忽略。

6.1 配置日志

默认情况下，Flask 的日志级别是 WARNING，只有警告和错误会输出。开发阶段可以设置为 DEBUG：

python

import logging

# 设置日志级别
app.logger.setLevel(logging.DEBUG)

更完整的项目通常使用dictConfig进行日志配置：

python

from logging.config import dictConfig

dictConfig({
    "version": 1,
    "formatters": {
        "default": {
            "format": "[%(asctime)s] %(levelname)s in %(module)s: %(message)s",
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stderr",
            "formatter": "default",
        },
    },
    "root": {
        "level": "INFO",
        "handlers": ["console"],
    },
})

app = Flask(__name__)

日志输出格式：

[2026-01-15 10:30:45] INFO in chat: 用户 user_123 发送了消息
[2026-01-15 10:30:46] ERROR in chat: Agent调用超时

6.2 在请求中记录日志

Agent API 中，可以记录每次请求的关键信息：

python

@app.route("/chat", methods=["POST"])
def chat():
    data = request.get_json()
    message = data.get("message", "")
    session_id = data.get("session_id", "default")

    app.logger.info(f"收到请求: session={session_id}, message={message[:50]}")

    try:
        # Agent处理逻辑...
        reply = f"收到: {message}"
    except Exception as e:
        app.logger.error(f"Agent处理失败: {e}", exc_info=True)
        abort(500)

    app.logger.info(f"回复完成: session={session_id}")
    return {"reply": reply}

exc_info=True会将完整异常堆栈写入日志，便于排查问题。

6.3 注入请求信息到日志

可以自定义 Formatter，使日志自动带上请求 URL 和 IP：

python

from flask import has_request_context, request


class RequestFormatter(logging.Formatter):
    def format(self, record):
        if has_request_context():
            record.url = request.url
            record.remote_addr = request.remote_addr
            record.method = request.method
        else:
            record.url = None
            record.remote_addr = None
            record.method = None

        return super().format(record)


formatter = RequestFormatter(
    "[%(asctime)s] %(remote_addr)s %(method)s %(url)s\n"
    "%(levelname)s in %(module)s: %(message)s"
)

日志输出示例：

[2026-01-15 10:30:45] 127.0.0.1 POST http://localhost:5000/api/chat
INFO in chat: 收到请求

这样每条日志都会带上请求 IP、方法和 URL，便于定位接口问题。

七、错误处理 + 日志组合

错误处理和日志通常需要结合使用。错误响应负责向调用方返回稳定格式，日志负责记录排查线索：

python

from flask import Flask, jsonify
from werkzeug.exceptions import HTTPException
import logging

app = Flask(__name__)


class AgentError(Exception):
    status_code = 400

    def __init__(self, message, status_code=None):
        super().__init__()
        self.message = message
        if status_code is not None:
            self.status_code = status_code


# 1. 自定义业务异常：记录warning日志，返回JSON
@app.errorhandler(AgentError)
def handle_agent_error(error):
    app.logger.warning(f"业务错误: {error.message}")
    return jsonify({"error": error.message, "code": error.status_code}), error.status_code


# 2. HTTP异常：返回JSON
@app.errorhandler(HTTPException)
def handle_http_exception(error):
    return jsonify({
        "error": error.description,
        "code": error.code,
    }), error.code


# 3. 未知异常：记录error日志（含堆栈），返回500
@app.errorhandler(Exception)
def handle_exception(error):
    if isinstance(error, HTTPException):
        return error

    app.logger.error(f"未处理异常: {error}", exc_info=True)
    return jsonify({"error": "服务器内部错误", "code": 500}), 500

该示例分为三层处理：

异常类型	日志级别	响应格式
`AgentError`（业务错误）	WARNING	JSON + 自定义消息
`HTTPException`（HTTP错误）	无	JSON + 标准描述
`Exception`（未知错误）	ERROR + 堆栈	JSON + "服务器内部错误"

八、生产环境日志建议

8.1 日志输出到文件

开发环境通常将日志输出到终端，生产环境通常需要写入文件：

python

from logging.config import dictConfig

dictConfig({
    "version": 1,
    "formatters": {
        "default": {
            "format": "[%(asctime)s] %(levelname)s in %(module)s: %(message)s",
        },
    },
    "handlers": {
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "filename": "logs/app.log",
            "maxBytes": 10 * 1024 * 1024,  # 10MB
            "backupCount": 5,
            "formatter": "default",
        },
    },
    "root": {
        "level": "INFO",
        "handlers": ["file"],
    },
})

RotatingFileHandler会自动轮转日志文件：超过 10MB 就新建文件，最多保留 5 个备份。

8.2 第三方库日志

Agent 项目中，LangChain、OpenAI 等库也会产生日志。可以为这些库单独设置日志级别：

python

# 给特定库设置日志级别
logging.getLogger("langchain").setLevel(logging.INFO)
logging.getLogger("openai").setLevel(logging.WARNING)

或者把所有库的日志都收集起来：

python

root = logging.getLogger()
root.setLevel(logging.INFO)

8.3 Sentry集成

生产环境中，也可以接入 Sentry 做错误监控。Sentry 可以自动捕获未处理异常、聚合重复错误，并发送告警：

bash

pip install sentry-sdk[flask]

python

import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration

sentry_sdk.init(
    dsn="your-sentry-dsn",
    integrations=[FlaskIntegration()],
)

配置完成后，导致 500 的异常会自动上报到 Sentry，并可在控制台查看堆栈、请求信息和发生频率。

九、总结

本篇主要介绍 API 在异常情况下保持稳定响应和可排查性的方式：

统一错误格式：@app.errorhandler()把所有错误转成JSON
自定义异常：AgentError携带业务错误码和详细信息
abort快速中断：参数校验失败时直接abort(400)
日志分级：DEBUG/INFO/WARNING/ERROR，按需输出
请求信息注入：每条日志带上IP、URL、方法
生产环境：日志写文件 + Sentry错误监控

统一错误处理和日志可以提高 Agent API 的稳定性和问题定位效率。

下一篇将介绍 Gunicorn + Nginx 部署，用于将 Agent API 从本地开发环境迁移到生产服务器。

06 错误处理与日志 ​

一、默认错误行为 ​

二、注册错误处理器 ​

2.1 按异常类注册 ​

2.2 捕获所有HTTP异常 ​

2.3 捕获未知异常 ​

三、自定义异常类 ​

四、abort函数 ​

五、Blueprint错误处理 ​

六、日志基础 ​

6.1 配置日志 ​

6.2 在请求中记录日志 ​

6.3 注入请求信息到日志 ​

七、错误处理 + 日志组合 ​

八、生产环境日志建议 ​

8.1 日志输出到文件 ​

8.2 第三方库日志 ​

8.3 Sentry集成 ​

九、总结 ​

06 错误处理与日志

一、默认错误行为

二、注册错误处理器

2.1 按异常类注册

2.2 捕获所有HTTP异常

2.3 捕获未知异常

三、自定义异常类

四、abort函数

五、Blueprint错误处理

六、日志基础

6.1 配置日志

6.2 在请求中记录日志

6.3 注入请求信息到日志

七、错误处理 + 日志组合

八、生产环境日志建议

8.1 日志输出到文件

8.2 第三方库日志

8.3 Sentry集成

九、总结